package mq;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import javax.annotation.PostConstruct;

/**
 * 
 * @author zhuyuanbiao
 *
 * @date 2017年11月17日 上午9:57:07
 */
public class RpcContext<T> {
	
	/** 监控请求是否超时 */
	private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
	
	private static ConcurrentMap<Long, ResponseFuture<?>> callbacks = new ConcurrentHashMap<>();
	
	/**
	 * 超时监控任务
	 * 
	 * @author zhuyuanbiao
	 *
	 * @date 2017年12月15日 下午5:34:07
	 */
	class TimeoutMonitor implements Runnable {

		@Override
		public void run() {
			callbacks.values().forEach((e) -> {
				if (e.isTimeout()) {
					Future<?> future = callbacks.remove(e.requestId());
					future.cancel();
				}
			});
		}
		
	}
	
	@PostConstruct
	private void start() {
		//executor.scheduleAtFixedRate(new TimeoutMonitor(), 0, 200, TimeUnit.MILLISECONDS);
	}
	
	/**
	 * 增加callback函数
	 * 
	 * @param future
	 */
	public static void addCallback(ResponseFuture<?> future) {
		callbacks.put(future.requestId(), future);
	}
	
	/**
	 * 执行回调函数
	 * 
	 * @param response
	 */
	public static void invokeCallback(Response response) {
		ResponseFuture<?> future = callbacks.remove(response.getRequestId());
		if (future == null) {
			return;
		}
		if (response.isError()) {
			future.onFailure(response);
			return;
		}
		future.onSuccess(response);
	}
	
}
